package de.preissler.dataimport;

import java.io.File;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.LinkedList;
import java.util.List;
import java.util.Scanner;
import java.util.StringTokenizer;

import me.prettyprint.cassandra.serializers.StringSerializer;
import me.prettyprint.hector.api.Cluster;
import me.prettyprint.hector.api.Keyspace;
import me.prettyprint.hector.api.beans.HColumn;
import me.prettyprint.hector.api.beans.HSuperColumn;
import me.prettyprint.hector.api.factory.HFactory;
import me.prettyprint.hector.api.mutation.Mutator;

/**
 * Importiert alle CSV Dateien
 * @author Eric Preissler
 *
 */
public class CSVImport
{
	private static final String OPEN = "open";
	private static final String HIGH = "high";
	private static final String LOW = "low";
	private static final String CLOSE = "close";
	private static final String VOLUME = "volume";
	private static final String CHANGE = "change";
	
	private Cluster cluster;
	private Keyspace keyspace;
	private String csvPath;
	
	public CSVImport(Cluster cluster, Keyspace keyspace, String csvPath)
	{
		this.cluster = cluster;
		this.keyspace = keyspace;
		this.csvPath = csvPath;
	}
	

	/**
	 * Startet den Kopiervorgang
	 * @throws Exception
	 */
	public void copyAllToCassandra() throws Exception
	{	
		//Pfade der auszulesenden Dateien besorgen
		List<String> filePathList = getFilePathList(csvPath);
		
		//Jede Datei einzeln auslesen
		for(String path : filePathList)
		{
			File file = new File(path);
			MyUtil.trace("Working on \"" + file.getName() + "\" (" + (filePathList.indexOf(path)+1) + "/" + filePathList.size() + ")!");
			
			copyFileToCassandra(cluster, keyspace, file);
		}
		
	}
	
	/*
	 * Gibt eine Liste zurueck mit dem Pfaden
	 */
	private List<String> getFilePathList(String csvPath)
	{
		File[] fileArray = new File(csvPath).listFiles();
		List<String> fileList = new LinkedList<String>();
		
		for(File f : fileArray)
		{
			if(f.isFile() && f.getName().endsWith(".csv"))
			{
				fileList.add(f.getPath());
			}
		}
		
		MyUtil.trace(fileList.size() + " .csv files has been found!");
		
		return fileList;
	}

	/*
	 * Kopiert eine .csv Datei zu Cassandra
	 */
	private void copyFileToCassandra(Cluster cluster, Keyspace keyspace, File file) throws Exception
	{	
		//Mutator holen, mit ihm werden Daten zu Cassandra geschrieben
		Mutator<String> mutator = HFactory.createMutator(keyspace, StringSerializer.get());

		//Scanner liest die Datei pro Zeile aus
		Scanner scanner = new Scanner(file);
		
		List<String> content = new LinkedList<String>();
		
		//Die Kopfzeile interessiert uns nicht
		scanner.next();
		
		//Gesamte Datei im RAM halten
		while(scanner.hasNext())
			content.add(scanner.next());
		
		//Rueckwaert auslesen damit der change value klappt
		//Jede Zeile ist eine SuperColumn
		double lastClose = 0;
		for(int index = content.size()-1; index >= 0; index--)
		{
			//Zeile in SuperColumn transferieren
			HSuperColumn<String, String, Double> sc = lineToSuperColumn(content.get(index), lastClose);
			lastClose = sc.getColumns().get(3).getValue();
			
			//Nun wird die SuperColumn unter der ISIN Zeile gespeichert
			mutator.insert(getIsinFromName(file.getName()), Main.SCF_QUOTES, sc);
		}
	}
	
	private HSuperColumn<String, String, Double> lineToSuperColumn(String line, double lastClose)
	{
		StringTokenizer tokenizer = new StringTokenizer(line, ";");
		
		List<HColumn<String, Double>> columnList = new ArrayList<HColumn<String, Double>>();
		
		//Datum
		String date = tokenizer.nextToken();
		Calendar calendar = Calendar.getInstance();
		StringTokenizer dateTokenizer = new StringTokenizer(date, "-");
		calendar.set(Integer.valueOf(dateTokenizer.nextToken()), Integer.valueOf(dateTokenizer.nextToken())-1,
				Integer.valueOf(dateTokenizer.nextToken()), 0, 0, 0);
		
		//Eroeffnung
		String open = tokenizer.nextToken();
		//Muss noch vom Deutschen Zahlenformat aufs Englische gebracht werden
		open = open.replace(".", "");
		open = open.replace(',', '.');
		columnList.add(HFactory.createColumn(OPEN, Double.valueOf(open), MyUtil.SS, MyUtil.DS));
		
		//Hoehste
		String high = tokenizer.nextToken();
		//Muss noch vom Deutschen Zahlenformat aufs Englische gebracht werden
		high = high.replace(".", "");
		high = high.replace(',', '.');
		columnList.add(HFactory.createColumn(HIGH, Double.valueOf(high), MyUtil.SS, MyUtil.DS));
		
		//Tiefster
		String low = tokenizer.nextToken();
		//Muss noch vom Deutschen Zahlenformat aufs Englische gebracht werden
		low = low.replace(".", "");
		low = low.replace(',', '.');
		columnList.add(HFactory.createColumn(LOW, Double.valueOf(low), MyUtil.SS, MyUtil.DS));
		
		//Schlusskurs
		String close = tokenizer.nextToken();
		//Muss noch vom Deutschen Zahlenformat aufs Englische gebracht werden
		close = close.replace(".", "");
		close = close.replace(',', '.');
		columnList.add(HFactory.createColumn(CLOSE, Double.valueOf(close), MyUtil.SS, MyUtil.DS));
		
		
		//Volumen
		String volume = tokenizer.nextToken();
		volume = volume.replace(".", "");
		volume = volume.replace(',', '.');
		columnList.add(HFactory.createColumn(VOLUME, Double.valueOf(volume), MyUtil.SS, MyUtil.DS));
		
		//Aenderung gegenueber Vortag
		double change = (Double.valueOf(close)-lastClose)/lastClose;
		columnList.add(HFactory.createColumn(CHANGE, change, MyUtil.SS, MyUtil.DS));
		

		return HFactory.createSuperColumn(date,
				columnList, MyUtil.SS, MyUtil.SS, MyUtil.DS);
	}
	
	/*
	 * Extrahiert die isin aus dem Dateinamen
	 */
	private static String getIsinFromName(String fileName)
	{
		return fileName.substring(fileName.indexOf('_')+1, fileName.lastIndexOf('_'));
	}

}
